package api.flinkapi.apidemos;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/7/25
 * 1,A,1627181351000
 * 1,C,1627181352000
 * 1,B,1627181353000
 * 1,C,1627181353010
 * 1,C,1627181354000
 * 1,X,1627181354001
 *
 * 1000注册timer
 * 1000+3000 = 4000 定时器执行点
 * 4001的时候，水印推进到4001，所以定时器被触发
 *
 **/
public class TimerDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> ds = env.socketTextStream("localhost", 5656);

        SingleOutputStreamOperator<Tuple3<String, String, Long>> dsTuple = ds
                .map(s -> Tuple3.of(s.split(",")[0], s.split(",")[1], Long.parseLong(s.split(",")[2])), TypeInformation.of(new TypeHint<Tuple3<String, String, Long>>() {
        }));
        WatermarkStrategy<Tuple3<String, String, Long>> assigner = WatermarkStrategy.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ofMillis(0)).withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
            @Override
            public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
                return element.f2;
            }
        });

        SingleOutputStreamOperator<Tuple3<String, String, Long>> waterStream = dsTuple.assignTimestampsAndWatermarks(assigner);

        KeyedStream<Tuple3<String, String, Long>, String> keyed = waterStream.keyBy(tp -> tp.f0);

        SingleOutputStreamOperator<String> res = keyed.process(new KeyedProcessFunction<String, Tuple3<String, String, Long>, String>() {
            ValueState<String> vs;
            @Override
            public void open(Configuration parameters) throws Exception {

                vs = getRuntimeContext().getState(new ValueStateDescriptor<String>("vs", String.class));

            }

            @Override
            public void processElement(Tuple3<String, String, Long> value, Context ctx, Collector<String> out) throws Exception {

                vs.update(value.f1);
                System.out.println(ctx.timestamp());
                if(value.f1.equals("A")){
                    System.out.println("注册了timer");
                    ctx.timerService().registerEventTimeTimer(value.f2 + 3000);
                }
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                System.out.println("ontime 被调用....");
                String now = vs.value();
                if(!now.equals("C")) {out.collect("A事件3秒后没有C");}else{out.collect("A事件3秒后发生C");}

            }
        });


        res.print();

        env.execute();
    }
}
